package io.mqttpush.mqttclient.service;

import io.mqttpush.common.events.CommonTopics;
import io.mqttpush.common.events.MsgEvent;
import io.mqttpush.common.events.OffLineEvent;
import io.mqttpush.common.events.OnLineEvent;
import io.mqttpush.common.events.utils.ByteBufEncodingUtil;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author tianzhenjiu
 */
public class DefaultMessageListener implements  MessageListener{


	Logger logger= LoggerFactory.getLogger(getClass());

	Map<String,Consumer> messageConsumers=new HashMap<>();


    Map<String,Consumer> eventConsumers=new HashMap<>();

    final String  endStart="*";

	@Override
	public void onRecivMessage(MqttPublishMessage mqttPublishMessage) {


		if(mqttPublishMessage==null){
			return;
		}


		MqttPublishVariableHeader mqttPublishVariableHeader=mqttPublishMessage.variableHeader();
		MqttFixedHeader mqttFixedHeader=mqttPublishMessage.fixedHeader();
		if(mqttFixedHeader.messageType()!= MqttMessageType.PUBLISH){
			logger.info("recv a not publish msg");
			return;
		}

		String topic=mqttPublishVariableHeader.topicName();



		if(topic==null){
            logger.info("recv a null topic  msg");
            return;
        }


        Map<String,Consumer> findConsumerMap=null;
		Object appectObj=null;
        ByteBuf byteBuf=mqttPublishMessage.content();
        if(topic.startsWith(CommonTopics.eventTopic)){



            if(topic.startsWith(CommonTopics.eventTopic.concat(OffLineEvent.class.getSimpleName()))){
                appectObj= ByteBufEncodingUtil.getInstance().decodeOffLineEvent(byteBuf);
            }
            else if(topic.startsWith(CommonTopics.eventTopic.concat(OnLineEvent.class.getSimpleName()))){
                appectObj= ByteBufEncodingUtil.getInstance().decodeOnLineEvent(byteBuf);
            }

            findConsumerMap=eventConsumers;


        }
        else{
            findConsumerMap=messageConsumers;
            appectObj=byteBuf;
        }




        if(findConsumerMap==null||findConsumerMap.isEmpty()){
            logger.info("can not find the consumer the map is empty");
            return;
        }

        if(findConsumerMap.containsKey(topic)){
            Consumer<Object> consumer=findConsumerMap.get(topic);
            if(consumer!=null&&appectObj!=null){
                consumer.accept(appectObj);
            }
            return;
        }

        final Object appectObjFal=appectObj;

        findConsumerMap.forEach((k,v)->{
            if(topic.startsWith(k)){
                v.accept(appectObjFal);
            }

        });





	}




    @Override
    public void addMessageProcesser(String topic, Consumer<ByteBuf> consumer) {


	    if(topic!=null&&consumer!=null){

	        if(topic.endsWith(endStart)){
	            topic=topic.substring(0,topic.length() - endStart.length());
            }

            messageConsumers.put(topic,consumer);
        }
    }

    @Override
    public void addEventProcesser(String topic, Consumer<MsgEvent> consumer) {
        if(topic!=null&&consumer!=null){


            if(topic.endsWith(endStart)){
                topic=topic.substring(0,topic.length() - endStart.length());
            }

            eventConsumers.put(topic,consumer);
        }
    }
}
